package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo01WordCnt {
  // 第一个Spark程序：wordCount
  def main(args: Array[String]): Unit = {
    // 构建Spark的配置文件
    val conf: SparkConf = new SparkConf()
    // 配置Spark的运行方式
    conf.setMaster("local")
    // 配置Spark的任务名称
    conf.setAppName("Demo01WordCnt")

    // 构建Spark环境
    val sc: SparkContext = new SparkContext(conf)

    /**
     * 环境构建好后开始写Spark代码
     */
    // 1、加载数据
    /**
     * RDD：弹性分布式数据集
     * 可以把它当成Scala中的集合使用
     * 实际上RDD是Spark中统一的编程模型
     * RDD中是不存储数据的
     */
    /**
     * RDD的五大特性：
     * 1、RDD是由一系列分区组成的，第一个RDD的分区数由切片的数量决定的
     * 2、Task是作用在每一个分区上的
     */
    val lineRDD: RDD[String] = sc.textFile("spark/data/words/*")
    println(lineRDD.getNumPartitions)

    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))


    val grpRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)
    println(grpRDD.getNumPartitions)


    val wordCntRDD: RDD[String] = grpRDD.map(kv => {
      println("进入map方法")
      s"${kv._1},${kv._2.size}"
    })
    println(wordCntRDD.getNumPartitions)

        wordCntRDD.foreach(println)
//    wordCntRDD.saveAsTextFile("spark/data/wcnt/")

  }

}
